package com.ruoyi.flink.dataStream;/**
 * @program: ruoyi
 * @description: OperatorTest
 * @author: zengchen
 * @create: 2020-12-06 15:58
 **/

import com.ruoyi.flink.source.CustomSourceFunction;
import com.ruoyi.flink.source.CustomSourceFunction1;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;


/**
 * @program: ruoyi
 * @description: 操作函数 练习
 * @author: zengchen
 * @create: 2020-12-06 15:58
 **/
public class OperatorTest {
    public static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        DataStreamSource<Long> dataStream = env.addSource(new CustomSourceFunction(), "nonPara2");
        DataStreamSource<Long> dataStream1 = env.addSource(new CustomSourceFunction1(), "nonPara21");

        KeyedStream<Tuple2<String, Long>, String> keyedStream = KeyByTuple2(env.addSource(new CustomSourceFunction(), "nonPara1"));

          // 1.dataStram
//        map(dataStream);
//        flatMap(dataStream);
//        filter(dataStream);
//        KeyBy(dataStream);

          // 2.keyedStream
        reduce(keyedStream); // 减少，两个相同key值的事件相互作用 只剩一个元素，从有两个相同的key开始
//        fold(keyedStream); // 折叠， 已废弃
//        sum(env); // 同 key元素指定属性相加起来,对于其他的字段取了最先获取的值
//        minBy(env); // minBy返回指定属性中具有最小值的整个元素，元素其它值也是这个元素本身的值
//        min(env); // min也会返回整个元素,不同的是min会根据指定的属性取最小值，并且把这个值保存在对应的位置上，
        // 对于其他的字段取了最先获取的值，不能保证每个元素的数值正确，max同理
//        max(env);
//        maxBy(env);

        // 3.window
//        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//        KeyedStream<Tuple4<String, Long, Long, String>, String> timestampStream = KeyByTuple4(dataStream);

//        window(timestampStream); // 根据某些特征将每个键中的数据分组（例如，在最近5秒钟内到达的数据）
//        windowAll(timestampStream); // 所有数据，one task
//        windowApply(timestampStream); //
//        windowReduce();  // 同 dataStream
//        windowFold();   // 同 dataStream
//        process(); // TODO 等元素全部到达之后再执行，执行之前数据将暂时存在内存里，区别于以前的方法，到达一条执行一下
//        min();// 同 dataStream
//        minBy();// 同 dataStream
//        max();// 同 dataStream
//        maxBy();// 同 dataStream
//        sum();// 同 dataStream

        // 4.
//        union(dataStream,dataStream1);
//        split_select(dataStream);
//        iterate(dataStream);
//        project(dataStream); // Only Tuple DataStreams can be projected

        // 5.物理分区
//        dataStream.partitionCustom(partitioner, "someKey");//自定义分区
//        dataStream.partitionCustom(partitioner, 0);//自定义分区
//        dataStream.shuffle(); // 随机分区
//        dataStream.rebalance(); // 重新平衡（轮询分区）
//        dataStream.rescale();//重新缩放
//        dataStream.broadcast(); // 将元素广播到每个分区。

        // 6.操作链
//        dataStream.filter(...).map(...).startNewChain().map(...);
//        dataStream.map(...).disableChaining();

        env.execute("operator test");
    }

    private static void project(DataStreamSource<Long> dataStream) {
        dataStream.map(new MapFunction<Long, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(Long value) throws Exception {
                if (value % 2 == 0) {
                    return new Tuple2<>("ou_shu", value);
                }
                return new Tuple2<>("ji_shu", value);
            }
        }).project(0).print();
    }

//    private static void split_select(DataStreamSource<Long> dataStream) {
//        SplitStream<Long> split = dataStream.split(new OutputSelector<Long>() {
//            @Override
//            public Iterable<String> select(Long value) {
//                List<String> output = new ArrayList<String>();
//                if (value % 2 == 0) {
//                    output.add("even");
//                } else {
//                    output.add("odd");
//                }
//                return output;
//            }
//        });
//        split.select("even").print();
//        split.select("odd").print();
//        split.select("even","odd").print();

//    }

    private static void union(DataStreamSource<Long> dataStream,DataStreamSource<Long> dataStream1) {
        dataStream.union(dataStream1).print();
    }

    private static void windowApply(KeyedStream<Tuple4<String, Long, Long, String>, String> timestampStream) {
        timestampStream.window(TumblingEventTimeWindows.of(Time.seconds(5))).apply(new WindowFunction<Tuple4<String, Long, Long, String>, Object, String, TimeWindow>() {
            @Override
            public void apply(String key, TimeWindow window, Iterable<Tuple4<String, Long, Long, String>> input, Collector<Object> out) throws Exception {
                if(input.iterator().hasNext()){ // 收集每5秒的第一个数据
                    out.collect(input.iterator().next());
                }
            }
        }).print();
    }

    private static void windowAll(KeyedStream<Tuple4<String, Long, Long, String>, String> timestampStream) {
        timestampStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5), Time.seconds(2))).minBy(1).print();
    }

    private static void window(KeyedStream<Tuple4<String, Long, Long, String>, String> timestampStream) {
        timestampStream.window(TumblingEventTimeWindows.of(Time.seconds(5))).minBy(1).print();
    }

    private static void minBy(StreamExecutionEnvironment env) {
        List<Tuple3<Integer, Integer, Integer>> data = new ArrayList<Tuple3<Integer, Integer, Integer>>();
        data.add(new Tuple3<>(0, 2, 2));
        data.add(new Tuple3<>(0, 1, 1));
        data.add(new Tuple3<>(0, 5, 6));
        data.add(new Tuple3<>(0, 3, 5));
        data.add(new Tuple3<>(1, 1, 9));
        data.add(new Tuple3<>(1, 2, 8));
        data.add(new Tuple3<>(1, 3, 10));
        data.add(new Tuple3<>(1, 2, 9));

        DataStreamSource<Tuple3<Integer, Integer, Integer>> items = env.fromCollection(data);
        items.keyBy(0).minBy(2).print();
    }

    private static void min(StreamExecutionEnvironment env) {
        List<Tuple3<Integer, Integer, Integer>> data = new ArrayList<Tuple3<Integer, Integer, Integer>>();
        data.add(new Tuple3<>(0, 2, 2));
        data.add(new Tuple3<>(0, 1, 1));
        data.add(new Tuple3<>(0, 5, 6));
        data.add(new Tuple3<>(0, 3, 5));
        data.add(new Tuple3<>(1, 1, 9));
        data.add(new Tuple3<>(1, 2, 8));
        data.add(new Tuple3<>(1, 3, 10));
        data.add(new Tuple3<>(1, 2, 9));

        DataStreamSource<Tuple3<Integer, Integer, Integer>> items = env.fromCollection(data);
        items.keyBy(0).min(2).print();
    }

    private static void sum(StreamExecutionEnvironment env) {
        List<Tuple3<Integer, Integer, Integer>> data = new ArrayList<Tuple3<Integer, Integer, Integer>>();
        data.add(new Tuple3<>(0, 2, 2));
        data.add(new Tuple3<>(0, 1, 1));
        data.add(new Tuple3<>(0, 5, 6));
        data.add(new Tuple3<>(0, 3, 5));
        data.add(new Tuple3<>(1, 1, 9));
        data.add(new Tuple3<>(1, 2, 8));
        data.add(new Tuple3<>(1, 3, 10));
        data.add(new Tuple3<>(1, 2, 9));

        DataStreamSource<Tuple3<Integer, Integer, Integer>> items = env.fromCollection(data);
        items.keyBy(0).sum(2).print();
    }

    // 已废弃
//    private static void fold(KeyedStream<Tuple2<String, Long>, String> keyedStream) {
//        keyedStream.fold()
//    }

    private static void reduce(KeyedStream<Tuple2<String, Long>, String> keyedStream) {
        keyedStream.reduce(new ReduceFunction<Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                System.out.println("value1:" + value1 + ",value2:" + value2);
                return Tuple2.of(value1.f0, value1.f1 + value2.f1);
            }
        }).print();
    }


    private static KeyedStream<Tuple2<String, Long>, String> KeyByTuple2(DataStreamSource<Long> nonPara) {
        return nonPara.map(new MapFunction<Long, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(Long value) throws Exception {
                if (value % 2 == 0) {
                    return new Tuple2<>("ou_shu", value);
                }
                return new Tuple2<>("ji_shu", value);
            }
        }).keyBy(value -> value.f0);
    }

    private static KeyedStream<Tuple4<String, Long, Long, String>, String> KeyByTuple4(DataStreamSource<Long> nonPara) {

        return nonPara.map(new MapFunction<Long, Tuple4<String, Long, Long, String>>() {
            @Override
            public Tuple4<String, Long, Long, String> map(Long value) throws Exception {
                LocalDateTime now = LocalDateTime.now();
                long nowTimestamp = now.toInstant(ZoneOffset.of("+8")).toEpochMilli();
                if (value % 2 == 0) {
                    return new Tuple4<>("ou_shu", value, nowTimestamp, now.format(dateTimeFormatter));
                }
                return new Tuple4<>("ji_shu", value, nowTimestamp, now.format(dateTimeFormatter));
            }
        }).assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple4<String, Long, Long, String>>() {
            @Nullable
            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(System.currentTimeMillis() - 1);
            }

            @Override
            public long extractTimestamp(Tuple4<String, Long, Long, String> element, long previousElementTimestamp) {
                return element.f2;
            }
        }).keyBy(value -> value.f0);
    }

    private static void filter(DataStreamSource<Long> nonPara) {
        nonPara.filter(value -> value % 2 == 0).print();
    }

    private static void flatMap(DataStreamSource<Long> nonPara) {
        nonPara.flatMap((FlatMapFunction<Long, Object>) (value, out) -> {
            if (value % 2 == 0) {
                out.collect(value);
            }
        }).print();
    }

    private static void map(DataStreamSource<Long> nonPara) {
        nonPara.map(x -> "so ease " + x).print();
    }

    public static KeyedStream<Tuple2<String, Long>, String> getKeyedStream(DataStreamSource<Long> nonPara) {
        return nonPara.map(new MapFunction<Long, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(Long value) throws Exception {
                if (value % 2 == 0) {
                    return new Tuple2<>("ou_shu", value);
                }
                return new Tuple2<>("ji_shu", value);
            }
        }).keyBy(value -> value.f0);
    }
}

